package com.example.springbootzk.utils;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description: curator 实现配置中心案例+常用api
 * @author: ZZZ
 * @time: 2021/4/6 16:14
 */
@Component
public class CuratorUtils {
    public static void main(String[] args) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.221.128:2181,192.168.221.129:2181,192.168.221.130:2181", 30 * 1000, 30 * 1000, retryPolicy);
        client.getState();
        client.start();
        CuratorFrameworkState state = client.getState();

        try {
            byte[] bytes = client.getData().forPath("/zk");
            System.out.println(new String(bytes));
        } catch (Exception exception) {
            exception.printStackTrace();
        }
    }


    private boolean zkFalg = false;
    //会话超时时间
    private final int SESSION_TIMEOUT = 30 * 1000;

    //连接超时时间
    private final int CONNECTION_TIMEOUT = 15 * 1000;

    @Value("${zk.url}")
    //ZooKeeper服务地址
    private String SERVER;

    //创建连接实例
    private CuratorFramework client = null;

    private ConcurrentHashMap<String, String> confmap = new ConcurrentHashMap<>();

    /**
     * baseSleepTimeMs：初始的重试等待时间
     * maxRetries：最多重试次数
     */
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 3);

    @PreDestroy
    private void close() {
        System.out.println("----curator 关闭-----");
        client.close();
    }

    @PostConstruct
    private void init() {
        if (!zkFalg) {
            synchronized (CuratorUtils.class) {
                if (!zkFalg) {
                    System.out.println("----curator 开始 初始化-----");
                    client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);
                    //为了避免多个应用的节点名称冲突的情况，CuratorFramework提供了命名空间的概念
                    //创建带命名空间的连接实例
//                    client = CuratorFrameworkFactory.builder()
//                            .namespace("zk/curator")
//                            .connectString(SERVER)
//                            .sessionTimeoutMs(SESSION_TIMEOUT)
//                            .connectionTimeoutMs(CONNECTION_TIMEOUT)
//                            .retryPolicy(retryPolicy)
//                            .build();

                    client.start();

                    System.out.println("----curator 初始化 完成-----");
                    //读取配置放入缓存 并且监听
                    try {
                        Stat stat = client.checkExists().forPath("/zk");
                        if(stat!=null){
                            byte[] bytes = client.getData().forPath("/zk");
                            confmap.put("/zk",new String(bytes));
                            NodeCache("/zk");

                            List<String> childrenList = client.getChildren().forPath("/zk");
                            if(childrenList!=null && childrenList.size()>0){
                                for (String key:childrenList) {
                                    String path="/zk/"+key;
                                    byte[] child = client.getData().forPath(path);
                                    confmap.put(path,new String(child));
                                }
                                PathChildrenCache("/zk");
                            }
                        }
                    } catch (Exception exception) {
                        exception.printStackTrace();
                    }
                }
            }
        }
    }

    //创建节点
    public String createNode(String path, String value) throws Exception {
        String result = client.create().withMode(CreateMode.PERSISTENT).forPath(path, value.getBytes());
        //client.create().creatingParentContainersIfNeeded().forPath(path,value.getBytes());
        //client.create().forPath(key,value.getBytes());
        //client.create().withMode(CreateMode.PERSISTENT).forPath(key,value.getBytes());
        // client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(key,value.getBytes());
        return result;
    }

    //设置值
    public Stat setData(String path, String value) throws Exception {
        Stat stat = client.setData().forPath(path, value.getBytes());
        //如果节点存在则Curator将会使用给出的数据设置这个节点的值，相当于 setData() 方法
        //client.create().orSetData().forPath(path, value.getBytes());
        //client.create().orSetData().creatingParentContainersIfNeeded().forPath(path,value.getBytes());
        return stat;
    }

    //获取值
    public String getData(String path) throws Exception {
        byte[] bytes = client.getData().forPath(path);
        return new String(bytes);
    }

    //删除
    public void deleteData(String path) throws Exception {
        client.delete().forPath(path);
        //级联删除子节点
        //client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
    }

    //获取子节点
    public List<String> getChildren(String path) throws Exception {
        List<String> list = client.getChildren().forPath(path);
        return list;
    }

    //节点是否存在
    public Boolean f(String path) throws Exception {
        Stat stat = client.checkExists().forPath(path);
        return stat == null ? false : true;
    }

    //监听
    public void dataListener() {
        client.getCuratorListenable().addListener(new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println(curatorEvent.getPath());
            }
        });
    }

    //NodeCache 监听一个节点，子节点也不关心
    public void NodeCache(String path) throws Exception {
        NodeCache nodeCache = new NodeCache(client, path);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                ChildData currentData = nodeCache.getCurrentData();
                if (currentData == null) {
                    System.out.println("节点:" + nodeCache.getPath() + " 被删除");
                    confmap.remove(nodeCache.getPath());
                } else {
                    System.out.println("节点:" + currentData.getPath() + " 发生变化 " + new String(currentData.getData()));
                    confmap.put(currentData.getPath(), new String(currentData.getData()));
                }
            }
        });
        nodeCache.start();
    }

    //监听子节点，孙子节点不监听
    public void PathChildrenCache(String path) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()){
                    case CHILD_REMOVED:{
                        confmap.remove(data.getPath());
                        break;
                    }
                    default:{
                        confmap.put(data.getPath(),new String(data.getData()));
                        break;
                    }
                }
            }
        });
        pathChildrenCache.start();
    }

    //监听所有子孙节点
    public void TreeCache(String path) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
                System.out.println(treeCacheEvent.getType().name());
            }
        });
        treeCache.start();

    }

    public ConcurrentHashMap<String, String> getConfmap() {
        return confmap;
    }

    public void setConfmap(ConcurrentHashMap<String, String> confmap) {
        this.confmap = confmap;
    }
}
